package com.shadow.disnet.netty.server;

import com.shadow.disnet.disrutor.DisruptorExecutor;
import com.shadow.disnet.disrutor.MessageWorker;
import com.shadow.disnet.netty.ILog;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.LinkedBlockingQueue;

public class MySeverInBoundHandler extends ChannelInboundHandlerAdapter implements ILog {

    private ChannelGroup group;
    private LinkedBlockingQueue<Channel> channels;
    private DisruptorExecutor executor;
    private NettyServer nettyServer;
    public final static AttributeKey<MessageWorker> WORKER_KEY = AttributeKey.valueOf("worker_key");

    public MySeverInBoundHandler(ChannelGroup group, LinkedBlockingQueue<Channel> channels, NettyServer nettyServer) {
        this.group = group;
        this.channels = channels;
        this.nettyServer  =nettyServer;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        info("入站通道激活");
        Channel channel = ctx.channel();
        if (group != null)
            this.group.add(channel);
        if (channels != null)
            this.channels.offer(channel);
        MessageWorker messageWorker = new MessageWorker(channel);
        this.executor = nettyServer.getExecutor(channel.id().hashCode());
        messageWorker.setExecutor(executor);
        channel.attr(WORKER_KEY).set(messageWorker); //将worker绑定起来
    }

    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        info("入站通道注册");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        info("入站通道关闭");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//        info("服務器读取到消息：%s", msg);
        String msgValue = (String) msg;
/*        executor.execute(() -> {
            info("服務器读取到消息：%s", msg);
        });*/
        MessageWorker messageWorker = ctx.channel().attr(WORKER_KEY).get();
        for (int i = 0; i < 100; i++) {
            executor.processMessage(1, 2, msgValue, messageWorker);
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        info("入站处理器读取消息完毕");

    }

    @Override
    public Logger getLogger() {
        return LoggerFactory.getLogger(this.getClass());
    }

    @Override
    public String toString() {
        return "入站处理器";
    }
}